#include "tcp_server.h"

#include "tcp_connection.h"
namespace zrpc {
TcpServer::TcpServer(EventLoop *loop, const std::string &ip, int16_t port,
                     const std::any &context)
    : loop_(loop),
      acceptor_(new Acceptor(loop, ip, port)),
      thread_pool_(new ThreadPool(loop)),
      context_(context) {
  acceptor_->SetNewConnectionCallback(
      std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
}

TcpServer::~TcpServer() {
  loop_->AssertInLoopThread();
  for (auto &it : connections_) {
    std::shared_ptr<TcpConnection> conn = it.second;
    it.second.reset();
    conn->GetLoop()->RunInLoop(
        std::bind(&TcpConnection::ConnectDestroyed, conn));
    conn.reset();
  }
}

void TcpServer::NewConnection(int32_t sockfd) {
  loop_->AssertInLoopThread();
  if (accept_callback_) {
    accept_callback_(sockfd);
  }

  EventLoop *loop = thread_pool_->GetNextLoop();
  std::shared_ptr<TcpConnection> conn(
      new TcpConnection(loop, sockfd, context_));
  connections_[sockfd] = conn;
  conn->SetConnectionCallback(std::move(connection_callback_));
  conn->SetMessageCallback(std::move(message_callback_));
  //conn->SetMessage1Callback(std::move(message1_callback_));
  conn->SetWriteCompleteCallback(std::move(write_complete_callback_));
  conn->SetCloseCallback(
      std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
  loop->RunInLoop(std::bind(&TcpConnection::ConnectEstablished, conn));
}

void TcpServer::SetThreadNum(int16_t num_threads) {
  thread_pool_->SetThreadNum(num_threads);
}

void TcpServer::Start() {
  thread_pool_->Start(thread_init_callback_);
  acceptor_->Listen();
}

void TcpServer::RemoveConnection(const std::shared_ptr<TcpConnection> &conn) {
  loop_->RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
}

void TcpServer::RemoveConnectionInLoop(
    const std::shared_ptr<TcpConnection> &conn) {
  loop_->AssertInLoopThread();
  size_t n = connections_.erase(conn->GetSockfd());
  (void)n;
  assert(n == 1);
  EventLoop *io_loop = conn->GetLoop();
  io_loop->QueueInLoop(std::bind(&TcpConnection::ConnectDestroyed, conn));
}
}  // namespace zrpc
